package com.practice.car.cardataapp.schelduer.spark.task.sale;

import com.alibaba.fastjson.JSON;
import com.practice.car.cardataapp.schelduer.spark.Schedule;
import com.practice.car.cardataapp.schelduer.spark.analy.SaleRunner;
import lombok.Data;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.*;
import scala.Serializable;
import scala.Tuple2;

import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import static com.practice.car.cardataapp.schelduer.spark.Schedule.getConnection;


/**
 * demo-study project
 *
 * @author FlagDied
 * @date 2020/8/27
 */
public class RddApplicaiton implements Serializable, SaleRunner {
	private Connection connection = null;

	@Override
	public void run() {
		String path = Schedule.saleHdfsUrl;
		//开启 SparkContext
		JavaSparkContext sc = new JavaSparkContext("local[2]", "rdd_application");
		JavaRDD<Car> carJavaRDD = sc.parallelize(readJsonArray(path, Car.class));
		everyMonthTop10(carJavaRDD);
		total_national_sales(carJavaRDD);
		car_sale_trend(carJavaRDD);
		national_car_sales_trends(carJavaRDD);
		suv_national_sales(carJavaRDD);
		sc.close();
	}


	public void everyMonthTop10(JavaRDD<Car> rdd) {
		//每月前10，这里直接统计了每个月的销量并排序了，如果要前10就拿前10的
		int year = new Date().getYear();

		String str = String.valueOf(year+1900);

		JavaPairRDD<String, Car> stringCarJavaPairRDD = rdd.filter(car->car.getDate().contains(str)).mapToPair(car -> new Tuple2<>(car.getDate(), car));
		stringCarJavaPairRDD.groupByKey().map(tuple2 -> {//要注意的是，在spark的各种遍历中，他给的都是序列化过的，如果你想要在
			//他的便利过程中使用其他的类，那么这个类需要实现 Serializeble
			Iterable<Car> cars = tuple2._2;
			List<Car> list = new ArrayList<>();
			cars.forEach(list::add);
			list.sort((a, b) -> b.getSale() - a.getSale());
			return new Tuple2<String, Iterable<Car>>(tuple2._1, list);
		}).foreach(tuple2 -> {// tuple2<String:date,Iterable<Car>:car_list>
			Iterable<Car> cars = tuple2._2;
			List<Car> carList=new ArrayList<>();
			cars.forEach(carList::add);
			int n = Math.min(carList.size(), 10);
			for (int i = 0; i < n; i++) {
				Car car = carList.get(i);
				PreparedStatement preparedStatement = getConnection().prepareStatement("insert into car_sale_month_top10(name,brand,month,sale) values(?,?,?,?)");
				preparedStatement.setString(1, car.getName());
				preparedStatement.setString(2, car.getBrand());
				preparedStatement.setString(3, car.getDate());
				preparedStatement.setInt(4,  car.getSale());
				preparedStatement.executeUpdate();
			}
		});

	}

	public void car_sale_trend(JavaRDD<Car> rdd) {
		JavaPairRDD<Tuple2<String,String>,Integer> carSaleTrend =rdd.mapToPair(car -> new Tuple2<>(new Tuple2<>(car.getType(), car.getDate()), (int)car.getSale()))
				.foldByKey(0, (a, b) -> a + b);
		carSaleTrend.foreachPartition(i -> i.forEachRemaining(item ->{
			Connection connection = getConnection();
			try {
				PreparedStatement preparedStatement = connection.prepareStatement("insert into car_sale_trend(type,month,sale) values(?,?,?)");
				preparedStatement.setString(1, item._1._1);
				//取小数点后三位，顺便改为百分比
				preparedStatement.setString(2, item._1._2);
				preparedStatement.setInt(3, item._2);
				preparedStatement.executeUpdate();
				preparedStatement.close();
			} catch (SQLException e) {
				e.printStackTrace();
			}
		}));
	}

	public void total_national_sales(JavaRDD<Car> rdd) {
		//国系总销量占比
		//映射为 Key-Value类型，Key为 national
		JavaPairRDD<String, Car> stringCarJavaPairRDD = rdd.mapToPair(car -> new Tuple2<>(car.getNational(), car));
		//对每个Key ：national 相同的Car的sale 进行聚合
		JavaPairRDD<String, Car> stringCarJavaPairRDD1 = stringCarJavaPairRDD.foldByKey(new Car(), (a, b) -> {
			b.setSale(a.getSale() + b.getSale());
			//这里如果用 a.setSale(a.getSale()+b.getSale()，那么结果的Car会是只有sale值，因为初始化的 new Car()只设置了 sale
			return b;
		});
		long sale = stringCarJavaPairRDD1.fold(new Tuple2<>("", new Car()), (a, b) -> {
			b._2.setSale(a._2.getSale() + b._2.getSale());
			return b;
		})._2.getSale();

		stringCarJavaPairRDD1.foreachPartition(i -> i.forEachRemaining(item -> {
			Connection connection = getConnection();
			try {
				PreparedStatement preparedStatement = connection.prepareStatement("insert into total_national_sales(national,proportion) values(?,?)");
				preparedStatement.setString(1, item._1);
				//取小数点后三位，顺便改为百分比
				preparedStatement.setDouble(2, (int) (item._2.getSale() * 1.0 / sale * 1000) / 10.0);
				preparedStatement.executeUpdate();
				preparedStatement.close();
			} catch (SQLException e) {
				e.printStackTrace();
			}
		}));
	}

	public void national_car_sales_trends(JavaRDD<Car> rdd) {
		//国系各月销量《=》国系销量走势
		JavaPairRDD<Tuple2<String, String>, Car> tuple2CarJavaPairRDD = rdd.mapToPair(car -> new Tuple2<>(new Tuple2<>(car.getNational(), car.getDate()), car));
		JavaPairRDD<Tuple2<String, String>, Car> tuple2CarJavaPairRDD1 = tuple2CarJavaPairRDD.foldByKey(new Car(), (a, b) -> {
			b.setSale(a.getSale() + b.getSale());
			return b;
		});
		tuple2CarJavaPairRDD1.foreachPartition(i -> i.forEachRemaining(item -> {
			Connection connection = getConnection();
			try {
				PreparedStatement preparedStatement = connection.prepareStatement("insert into national_car_sales_trends(national,month,sale) values(?,?,?)");
				preparedStatement.setString(1, item._1._1);
				//取小数点后三位，顺便改为百分比
				preparedStatement.setString(2, item._1._2);
				preparedStatement.setInt(3, (int)item._2.getSale());
				preparedStatement.executeUpdate();
				preparedStatement.close();
			} catch (SQLException e) {
				e.printStackTrace();
			}
		}));
	}

	private void suv_national_sales(JavaRDD<Car> rdd) {
		//SUV 各国系总销量对比，这里直接统计各国各车型的总销量
		JavaPairRDD<String,Car> tuple2CarJavaPairRDD = rdd.mapToPair(car -> new Tuple2<>(car.getNational(), car))
				.foldByKey(new Car(), (a, b) -> {
					a.setSale(a.getSale() + b.getSale());
					return a;
				});
		tuple2CarJavaPairRDD.foreachPartition(i -> i.forEachRemaining(item -> {
			Connection connection = getConnection();
			try {
				PreparedStatement preparedStatement = connection.prepareStatement("insert into suv_national_sales(national,sales) values(?,?)");
				preparedStatement.setString(1, item._1);
				preparedStatement.setInt(2, (int)item._2.getSale());
				preparedStatement.executeUpdate();
				preparedStatement.close();
			} catch (SQLException e) {
				e.printStackTrace();
			}
		}));
	}


	private <T> List<T> readJsonArray(String path, Class<T> clasz) {
		List<T> list = null;
		int available = 0;
		Path hdfsPath = new Path(path);
		Configuration configuration = new Configuration();
		FSDataInputStream fsDataInputStream = null;
		FileSystem fileSystem = null;
		try {
			fileSystem =hdfsPath.getFileSystem(configuration);
			fsDataInputStream = fileSystem.open(hdfsPath);
			BufferedInputStream inputStream = new BufferedInputStream(fsDataInputStream);
			available = inputStream.available();
			byte[] buffer = new byte[available];
			inputStream.read(buffer);
			inputStream.close();
			list = JSON.parseArray(new String(buffer, StandardCharsets.UTF_8), clasz);
		} catch (IOException e) {
			e.printStackTrace();
		}
		//将 text文件里的 json数据读取 List<Car>
		return list;
	}


	@Data
	public static class Car implements Serializable {
		String name;
		String brand;
		String type;
		double price;
		String date;
		int sale;
		String national;

		public Car() {
			name = "";
			brand = "";
			type = "";
			price = 0;
			date = "";
			sale = 0;
			national = "";
		}

	}
}
